package cn.itcast.producer;

import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.Properties;

/**
 * @author lilulu
 * @date 2023-04-16 20:19
 */
public class KafkaMsgProducer implements Runnable {
    private Logger logger = LoggerFactory.getLogger(KafkaMsgProducer.class);

    public static void main(String[] args) {
        new Thread(new KafkaMsgProducer()).start();
    }

    @SneakyThrows
    @Override
    public void run() {
        KafkaProducer kafkaProducer = createKafkaProducer();

        logger.info("开始生产信息！！！！");
        BufferedReader bufferedReader = null;
        try {
            File file = new File("E:\\vehicleProject\\CarNetworkingSystem\\data\\sourcedata.txt");
            bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
            String line;
            int count = 1;
            while ((line = bufferedReader.readLine()) != null) {
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("vehicle-data-json", line);
                kafkaProducer.send(record);
                System.out.println("模拟数据发送程序，生产第"+count+"条数据");
                count ++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (bufferedReader != null) {
                bufferedReader.close();
            }
            kafkaProducer.close();
        }
    }

    public KafkaProducer createKafkaProducer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("request.required.acks", "1");
        properties.put("producer.type", "async");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        return kafkaProducer;
    }
}
